package com.ks.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.ks.common.PageResult;
import com.ks.vo.ReceiveVO;
import com.ks.dto.MetricDataDTO;
import com.ks.dto.PageQueryDTO;
import com.ks.entity.MetricData;
import com.ks.exception.IllegalConfigurationException;
import com.ks.mapper.MetricDataMapper;
import com.ks.service.MetricDataService;
import com.ks.service.WarnRecordService;
import com.ks.vo.MetricDataVO;
import com.ks.vo.RequestVO;
import jdk.nashorn.internal.ir.RuntimeNode;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;

import javax.annotation.PostConstruct;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;


@Service
@Slf4j
public class MetricDataServiceImpl extends ServiceImpl<MetricDataMapper, MetricData> implements MetricDataService {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private WarnRecordService warnRecordService;

    @Autowired
    private PlatformTransactionManager transactionManager;

    @Autowired
    private MetricDataMapper metricDataMapper;

    @Value("${config.mode}")
    private String mode;


    private static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors(); // 核心线程数为CPU核心数
    private static final int MAXIMUM_POOL_SIZE = CORE_POOL_SIZE * 2; // 最大线程数为核心线程数的两倍
    private static final long KEEP_ALIVE_TIME = 300L; // 临时线程最大存活时间
    private static final BlockingQueue<Runnable> WORK_QUEUE = new LinkedBlockingQueue<>(); // 线程工厂任务队列
    private static final ThreadFactory THREAD_FACTORY = Executors.defaultThreadFactory(); // 默认线程工厂
    private static final RejectedExecutionHandler HANDLER = new ThreadPoolExecutor.AbortPolicy(); // 默认拒绝策略

    private final ThreadPoolExecutor THREAD_POOL_EXECUTOR = new ThreadPoolExecutor(
            CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.MILLISECONDS,
            WORK_QUEUE, THREAD_FACTORY, HANDLER);

    private final BlockingQueue<List<MetricData>> TASKS = new ArrayBlockingQueue<>(16 * 1024 * 1024); // jdk阻塞队列,没采用redis stream
    private static final ExecutorService DATABASE_EXECUTOR = Executors.newSingleThreadExecutor(); // 单线程工厂

    @Override
    public ReceiveVO dealMonitorData(List<MetricDataDTO> metricDataDTOList) {
        // 将监控数据按时间戳分类
        Map<Long, List<MetricDataDTO>> classifiedData = classifyByTimestamp(metricDataDTOList);

        // 对每个时间戳的数据进行处理
        for (Map.Entry<Long, List<MetricDataDTO>> entry : classifiedData.entrySet()) {
            Long timestamp = entry.getKey();
            List<MetricDataDTO> dataForTimestamp = entry.getValue();
            // 调用 dealDataForTimestamp 方法处理相同时间戳的数据
            THREAD_POOL_EXECUTOR.submit(() -> dealDataForTimestamp(timestamp, dataForTimestamp));
        }

        // 封装返回数据
        ReceiveVO receiveVO = new ReceiveVO();
        receiveVO.setMessage("ok");
        receiveVO.setRequestId(UUID.randomUUID().toString());
        return receiveVO;
    }

    private Map<Long, List<MetricDataDTO>> classifyByTimestamp(List<MetricDataDTO> metricDataDTOList) {
        // 根据时间戳分类监控数据
        Map<Long, List<MetricDataDTO>> classifiedData = new HashMap<>();
        for (MetricDataDTO metricDataDTO : metricDataDTOList) {
            Long timestamp = metricDataDTO.getTimestamp().longValue();
            classifiedData.computeIfAbsent(timestamp, k -> new ArrayList<>()).add(metricDataDTO);
        }
        return classifiedData;
    }

    private void dealDataForTimestamp(Long timestamp, List<MetricDataDTO> dataForTimestamp) {
        // 在这里处理相同时间戳的监控数据
        List<MetricData> metricDataList = convertMetricDataDTOList(dataForTimestamp);
        // 将 MetricData 保存到队列,另一线程会不断读取和保存到数据库
        TASKS.add(metricDataList);
        // 把监控数据存入redis zset,score为系统当前时间,供定时任务获取近期监控数据
        if ("bySystemCurrentTime".equals(mode)) {
            stringRedisTemplate.opsForZSet().add("metricData", JSON.toJSONString(metricDataList), System.currentTimeMillis());
        } else if ("byJsonDataTime".equals(mode)) {
            stringRedisTemplate.opsForZSet().add("metricData", JSON.toJSONString(metricDataList), timestamp * 1000);
        } else {
            throw new IllegalConfigurationException("不存在的模式配置");
        }

        // 异步调用实时计算
        warnRecordService.currentCalculate(metricDataList);
    }

    @PostConstruct
    private void saveTask() {
        DATABASE_EXECUTOR.submit(() -> {
            List<MetricData> metricDataList = null;
            while (true) {
                TransactionStatus status = transactionManager.getTransaction(new DefaultTransactionDefinition());
                try {
                    metricDataList = TASKS.take(); //从阻塞队列获取待写入数据库的监控数据
                    saveBatch(metricDataList);
                    transactionManager.commit(status);
                } catch (Exception e) {
                    transactionManager.rollback(status);
                    log.error("保存监控数据至数据库失败,数据为{}", metricDataList);
                    e.printStackTrace();
                }
            }
        });
    }


    private List<MetricData> convertMetricDataDTOList(List<MetricDataDTO> metricDataDTOList) {
        return metricDataDTOList.stream()
                .map(this::convertMetricDataDTO)
                .collect(Collectors.toList());
    }

    private MetricData convertMetricDataDTO(MetricDataDTO metricDataDTO) {
        MetricData metricData = new MetricData();
        metricData.setMetric(metricDataDTO.getMetric());
        metricData.setInstanceId(metricDataDTO.getTags().get("instanceId"));
        metricData.setHostname(metricDataDTO.getTags().get("hostname"));
        Instant instant = Instant.ofEpochSecond(metricDataDTO.getTimestamp());
        LocalDateTime localDateTime = instant.atZone(ZoneId.systemDefault()).toLocalDateTime();
        metricData.setTimestamp(localDateTime);
        metricData.setValue(metricDataDTO.getValue());
        return metricData;
    }


    @Override
    public PageResult pageMetricData(PageQueryDTO pageQueryDTO) {
        IPage page = new Page(pageQueryDTO.getPage(), pageQueryDTO.getPageSize());
        LambdaQueryWrapper<MetricData> lambdaQueryWrapper = new LambdaQueryWrapper<>();
        page = page(page, lambdaQueryWrapper);
        PageResult pageResult = PageResult.builder()
                .total(page.getTotal())
                .records(page.getRecords())
                .build();
        return pageResult;
    }


    @Override
    public PageResult pageChartData(PageQueryDTO pageQueryDTO, String metric) {
        IPage page = new Page(pageQueryDTO.getPage(), pageQueryDTO.getPageSize());
        LambdaQueryWrapper<MetricData> lambdaQueryWrapper = new LambdaQueryWrapper<>();
        lambdaQueryWrapper.eq(MetricData::getMetric, metric);
        page = page(page, lambdaQueryWrapper);
        PageResult pageResult = PageResult.builder()
                .total(page.getTotal())
                .records(page.getRecords())
                .build();
        return pageResult;
    }

    @Override
    public List<MetricDataVO> queryMetricDataByTime(RequestVO requestVO) {

        LocalDateTime startTime = convertStringToDatetime(requestVO.getStart());
        LocalDateTime endTime = convertStringToDatetime(requestVO.getEnd());

        // 查询数据库
        List<MetricData> metricDataList = metricDataMapper.selectMetricDataByTime(startTime, endTime);

        // 转换为 VO 对象
        return metricDataList.stream()
                .map(MetricDataConverter::toVO)
                .collect(Collectors.toList());
    }

    //把String类型的时间转化成Datetime类型
    public static LocalDateTime convertStringToDatetime(String time) {
        final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        try {
            return LocalDateTime.parse(time, FORMATTER);
        } catch (DateTimeParseException e) {
            // 处理解析异常，例如记录日志或抛出自定义异常
            throw new IllegalArgumentException("非法的时间格式, 应为 'yyyy-MM-dd HH:mm:ss'", e);
        }
    }

    // 把 MetricData 转换为 VO 对象
    private static class MetricDataConverter {
        public static MetricDataVO toVO(MetricData metricData) {
            MetricDataVO metricDataVO = new MetricDataVO();
            metricDataVO.setMetric(metricData.getMetric());
            // 将 LocalDateTime 转换为时间戳并赋值
            metricDataVO.setTimestamp((int) metricData.getTimestamp().toEpochSecond(ZoneOffset.UTC));
            metricDataVO.setValue(metricData.getValue());
            return metricDataVO;
        }
    }
}
